package com.xiaohu.transfrom.streamtranform;

import com.xiaohu.transfrom.MyPartitioner;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/*
    分流

    需求：将数据按照奇数和偶数分开为两个流进行处理
 */
public class FilterDemo {
    public static void main(String[] args) throws Exception{
        Configuration conf = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        env.setParallelism(2);
        //设置流处理环境还是批处理环境 DataSet API已经过时了，现在都是一套代码，进行设置
//        env.setRuntimeMode(RuntimeExecutionMode.BATCH); //批处理
//        env.setRuntimeMode(RuntimeExecutionMode.STREAMING); //流处理，默认就是流处理
        //一般情况下，不会在代码中指定，不够灵活，一般都是在提交的时候，使用命令进行指定 flink run  -Dexecution.runtime-mode=BATCH【STREAMING】 ...

        DataStreamSource<String> socketDS = env.socketTextStream("master", 7777);

        //可以使用两遍filter算子，分别过滤奇数或偶数
        //缺点：socketDS被处理了两次
        socketDS.filter(e->Integer.parseInt(e)%2==0).print("偶数流：");
        socketDS.filter(e->Integer.parseInt(e)%2==1).print("奇数流：");

        //使用侧输出流



        env.execute();
    }
}
